package com.Reactor3参考文档;

import org.junit.Test;
import org.reactivestreams.Subscription;

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class Subscribe {

	@Test
	public void test1(){
		// 配置一个在订阅时会产生3个值的 Flux。
		Flux<Integer> ints = Flux.range(1, 3);
		// 最简单的订阅方式。
		// 第二行代码没有任何输出，但是它确实执行了。Flux 产生了3个值。如果我们传入一个 lambda， 我们就可以看到这几个值，如下一个列子：
		ints.subscribe();
	}

	@Test
	public void test2(){
		Flux<Integer> ints = Flux.range(1, 3);
		ints.subscribe(i -> System.out.println(i));
	}

	@Test
	// 为了演示下一个方法签名，我们故意引入一个错误，如下所示：
	public void test3(){
		// 配置一个在订阅时会产生4个值的 Flux。
		Flux<Integer> ints = Flux.range(1, 4)
				// 为了对元素进行处理，我们需要一个 map 操作。
		      .map(i -> {
		    	  // 对于多数元素，返回值本身。
		        if (i <= 3) return i;
		        // 对其中一个元素抛出错误。
		        throw new RuntimeException("Got to 4");
		      });
		ints.subscribe(i -> System.out.println(i),
				// 订阅的时候定义如何进行错误处理。
		      error -> System.err.println("Error: " + error));
	}

	@Test
	// 下一个 subscribe 方法的签名既有错误处理，还有一个完成后的处理，如下：
	// 错误和完成信号都是终止信号，并且二者只会出现其中之一。为了能够最终全部正常完成，你必须处理错误信号。
	// 用于处理完成信号的 lambda 是一对空的括号，因为它实际上匹配的是 Runnalbe 接口中的 run 方法， 不接受参数。
	public void test4(){
		Flux<Integer> ints = Flux.range(1, 4);
		ints.subscribe(i -> System.out.println(i),
		    error -> System.err.println("Error " + error),
		    () -> {System.out.println("Done");});
	}

	/**
	 * SampleSubscriber 类继承自 BaseSubscriber，在 Reactor 中, 推荐用户扩展它来实现自定义的 Subscriber。这个类提供了一些 hook 方法，我们可以通过重写它们来调整 subscriber 的行为。 默认情况下，它会触发一个无限个数的请求，但是当你想自定义请求元素的个数的时候，扩展 BaseSubscriber 就很方便了。
扩展的时候通常至少要覆盖 hookOnSubscribe(Subscription subscription) 和 hookOnNext(T value) 这两个方法。这个例子中， hookOnSubscribe 方法打印一段话到标准输出，然后进行第一次请求。 然后 hookOnNext 同样进行了打印，同时逐个处理剩余请求。
SampleSubscriber 输出如下：
Subscribed
1
2
3
4
 	建议你同时重写 hookOnError、hookOnCancel，以及 hookOnComplete 方法。 你最好也重写 hookFinally 方法。SampleSubscribe 确实是一个最简单的实现了 请求有限个数元素的 Subscriber。
	 */
	@Test
	public void test5(){
		SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
		Flux<Integer> ints = Flux.range(1, 4);
		ints.subscribe(i -> System.out.println(i),
		    error -> System.err.println("Error " + error),
		    () -> {System.out.println("Done");},
		    s -> ss.request(10));
		ints.subscribe(ss);
	}

	static class SampleSubscriber<T> extends BaseSubscriber<T> {

        public void hookOnSubscribe(Subscription subscription) {
                System.out.println("Subscribed");
                request(1);
        }

        public void hookOnNext(T value) {
                System.out.println(value);
                request(1);
        }
	}

	@Test
	// 响应式流规范定义了另一个 subscribe 方法的签名，它只接收一个自定义的 Subscriber， 没有其他的参数，如下所示：
	public void test6() {
		Flux<String> source = someStringSource();

		//BaseSubscriber 定义了多种用于处理不同信号的 hook。它还定义了一些捕获 Subscription 对象的现成方法，这些方法可以用在 hook 中。
	 	//request(n) 就是这样一个方法。它能够在任何 hook 中，通过 subscription 向上游传递 背压请求。
		//这里我们在开始(hookOnSubscribe)这个流的时候请求1个元素值。
	 	//随着接收到新的值(hookOnNext)，我们继续以每次请求一个元素的节奏从源头请求值。
	 	//其他 hooks 有 hookOnComplete, hookOnError, hookOnCancel, and hookFinally （它会在流终止的时候被调用，传入一个 SignalType 作为参数）。
		//当你修改请求操作的时候，你必须注意让 subscriber 向上提出足够的需求， 否则上游的 Flux 可能会被“卡住”。所以 BaseSubscriber 在进行扩展的时候要覆盖 hookOnSubscribe 和 onNext，这样你至少会调用 request 一次。
		//BaseSubscriber 还提供了 requestUnbounded() 方法来切换到“无限”模式（等同于 request(Long.MAX_VALUE)）
		source.map(String::toUpperCase).subscribe(new BaseSubscriber<String>() {
			@Override
			protected void hookOnSubscribe(Subscription subscription) {
				request(1);
			}

			@Override
			protected void hookOnNext(String value) {
				request(1);
			}

		});
	}

	private Flux<String> someStringSource() {
		return Flux.just("1", "2", "3");
	}

}
